package cn.itcast.tags.models.ml

import cn.itcast.tags.models.{AbstractModel, ModelType}
import cn.itcast.tags.tools.{MLModelTools, TagTools}
import org.apache.spark.ml.clustering.KMeansModel
import org.apache.spark.ml.feature.VectorAssembler
import org.apache.spark.sql.{Column, DataFrame, SparkSession}
import org.apache.spark.sql.functions._
import org.apache.spark.sql.types.DataTypes

/**
 * 挖掘类型标签：PSM 价格敏感度模型
 */
class PsmModel extends AbstractModel("价格敏感度标签", ModelType.ML) {
  /*
    372	消费敏感度
      373	极度敏感		  0
      374	比较敏感		  1
      375	一般敏感		  2
      376	不太敏感		  3
      377	极度不敏感		4
  */
  override def doTag(businessDF: DataFrame, tagDF: DataFrame): DataFrame = {
    val spark: SparkSession = businessDF.sparkSession
    import spark.implicits._
    /*
      TODO: 1. 按照业务数据计算每个用户的PSM值
        psm = 优惠订单占比（tdonr） + 平均优惠金额占比（adar） + 优惠总金额占比（tdar）
        tdonr = 优惠订单占比(优惠订单数 / 订单总数)
        adar = 平均优惠金额占比(平均优惠金额 / 平均每单应收金额)
        tdar = 优惠总金额占比(优惠总金额 / 订单总金额)
     */
    // ra: receivableAmount 应收金额
    val raColumn: Column = ($"orderamount" + $"couponcodevalue").as("ra")
    // da: discountAmount 优惠金额
    val daColumn: Column = $"couponcodevalue".cast(DataTypes.createDecimalType(10, 2)).as("da")
    // pa: practicalAmount 实收金额
    val paColumn: Column = $"orderamount".cast(DataTypes.createDecimalType(10, 2)).as("pa")
    // state: 订单状态，此订单是否是优惠订单
    val stateColumn: Column = when($"couponcodevalue" === 0.0, 0)
      .otherwise(1).as("state")

    // tdon 优惠订单数
    val tdonColumn: Column = sum($"state").as("tdon")
    // ton 总订单总数
    val tonColumn: Column = count($"state").as("ton")
    // tda 优惠总金额
    val tdaColumn: Column = sum($"da").as("tda")
    // tra 应收总金额
    val traColumn: Column = sum($"ra").as("tra")

    /*
      tdonr 优惠订单占比(优惠订单数 / 订单总数)
      tdar 优惠总金额占比(优惠总金额 / 订单总金额)
      adar 平均优惠金额占比(平均优惠金额 / 平均每单应收金额)
    */
    val tdonrColumn: Column = ($"tdon" / $"ton").as("tdonr")
    val tdarColumn: Column = ($"tda" / $"tra").as("tdar")
    val adarColumn: Column = (($"tda" / $"tdon") / ($"tra" / $"ton")).as("adar")
    val psmColumn: Column = ($"tdonr" + $"tdar" + $"adar").as("psm")

    val psmDF: DataFrame = businessDF
      // 确定每个订单ra、da、pa及是否为优惠订单
      .select(
        $"memberid".as("userId"),
        raColumn, daColumn, paColumn, stateColumn
      )
      // 按照userId分组，聚合统计：订单总数和订单总额
      .groupBy($"userId")
      .agg(
        tonColumn, tdonColumn, traColumn, tdaColumn
      )
      // 计算优惠订单占比、优惠总金额占比、adar
      .select(
        $"userId", tdonrColumn, tdarColumn, adarColumn
      )
      // 计算PSM值
      .select($"*", psmColumn)
      .select(
        $"*", //
        when($"psm".isNull, 0.00000001)
          .otherwise($"psm").as("psm_score")
      )
    // psmDF.printSchema()
    // psmDF.show(50, truncate = false)

    // TODO: 2. 使用KMeans聚类算法，训练模型，针对psm单列值聚类操作
    // 2.1 特征值features
    val psmFeaturesDF: DataFrame = new VectorAssembler()
      .setInputCols(Array("psm_score"))
      .setOutputCol("features")
      .transform(psmDF)
    // 2.2 获取KMeans模型
    val kMeansModel: KMeansModel = MLModelTools
      .loadModel(psmFeaturesDF, "psm", this.getClass).asInstanceOf[KMeansModel]

    // TODO: 3. 模型预测评估，和属性标签规则打标签
    // 3.1 模型预测
    val predictionDF: DataFrame = kMeansModel.transform(psmFeaturesDF)
    // predictionDF.printSchema()
    // predictionDF.show(50, truncate = false)

    // 3.2 打标签
    val modelDF: DataFrame = TagTools.kMeansMatchTag(kMeansModel, predictionDF, tagDF, "psm")
    modelDF.show(100, truncate = false)
    // 返回标签数据
    modelDF
  }
}

object PsmModel {
  def main(args: Array[String]): Unit = {
    val tagModel = new PsmModel()
    tagModel.executeModel(372L)
  }
}
